package com.shujia.dim

import com.shujia.common.utils.SparkMain
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DimScenicGrid extends SparkMain {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 加载景区边界表、网格配置表
    val scenicBoundaryDF: DataFrame = spark.table("dim.dim_scenic_boundary")
    val geoTagGridDF: DataFrame = spark.table("dim.dim_geotag_grid").select($"grid_id")


    /**
     * 同样没有关联条件 所以直接通过笛卡尔积关联
     * 由于两张表的数据基本上都是不更新的 所以只需要做一次笛卡尔积即可 不需要每天调度
     * 所以不会对性能造成影响
     */

    geoTagGridDF
      .crossJoin(scenicBoundaryDF)
      // 通过网格id以及景区边界 判断网格在不在景区内部
      .where(isInBoundaryWithGrid($"boundary", $"grid_id"))
      // 将属于同一个景区内的所有的网格合并成一条记录并转成字符串
      .groupBy($"scenic_id", $"scenic_name")
      .agg(concat_ws("#", collect_set($"grid_id")) as "grids")
      // 保存数据
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save("/daas/motl/dim/dim_scenic_grid")


  }
}
